package com.leilei;

import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Iterator;
import java.util.Random;
import java.util.UUID;

/**
 * @author lei
 * @version 1.0
 * @desc flink 有状态计算  mapState使用
 * @date 2021-03-19 14:36
 */
public class Flink_State_2_MapState {
    public static void main(String[] args) throws Exception {
        //开启本地web-ui
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 8082);
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        env.setParallelism(4);
        final DataStreamSource<Location> locationSource = env.addSource(new LocationSource());
        final WindowedStream<Location, Integer, GlobalWindow> windowedStream = locationSource
                .keyBy(Location::getVehicleId)
                .countWindow(1);
        DataStream<String> result = windowedStream.apply(new SpeedAlarmWindow());
        result.print();
        //result.addSink(new MysqlSink());
        env.execute("mapState-window");

    }


    public static class SpeedAlarmWindow extends RichWindowFunction<Location, String, Integer, GlobalWindow> {
        MapState<String, Location> locationState;
        String uuid;

        @Override
        public void apply(Integer integer, GlobalWindow window, Iterable<Location> locationList, Collector<String> out) throws Exception {
            for (Location location : locationList) {
                final String key = location.getVehicleId().toString();
                final Location preLocation = locationState.get(key);
                if (preLocation == null) {
                    if (location.getGpsSpeed() > location.getLimitSpeed()) {
                        locationState.put(key, location);
                        out.collect(location.toString() + ">>超速开始");
                        break;
                    }
                } else {
                    if (location.getGpsSpeed() > location.getLimitSpeed()) {
                        locationState.put(key, location);
                        out.collect(location.toString() + ">>持续超速中" +  "\n" +
                                "上一条超速数据为：" + "\n" + preLocation.toString());
                    } else {
                        locationState.remove(key);
                        out.collect(location.toString() + ">>超速结束" + "\n"+" 结束前定位:" + "\n" + preLocation.toString());
                    }
                }
            }
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            uuid = UUID.randomUUID().toString();
            System.out.println(uuid + "窗口打开了");
            super.open(parameters);
            // 定义状态描述器
            locationState = getRuntimeContext().getMapState(new MapStateDescriptor<>("locationState",
                    TypeInformation.of(String.class),
                    TypeInformation.of(Location.class)));

        }

        @Override
        public void close() throws Exception {
            super.close();
            System.out.println(uuid + "窗口关闭了");
        }
    }

    public static class LocationSource implements SourceFunction<Location> {
        Boolean flag = true;
        @Override
        public void run(SourceContext<Location> ctx) throws Exception {
            Random random = new Random();
            while (flag) {
                int vehicleId = random.nextInt(20) + 1;
                Location location = Location.builder()
                        .vehicleId(vehicleId)
                        .plate("川A000" + vehicleId)
                        .color("绿")
                        .date(Integer.parseInt(LocalDate.now().format(DateTimeFormatter.BASIC_ISO_DATE)))
                        .gpsSpeed(RandomUtil.randomInt(88, 100))
                        .limitSpeed(RandomUtil.randomInt(88, 95))
                        .devTime(System.currentTimeMillis())
                        .build();
                ctx.collect(location);
                Thread.sleep(2000);
            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
    }

    /**
     * 自定义mysql sink
     */
    public static class MysqlSink extends RichSinkFunction<String> {
        Connection conn = null;
        PreparedStatement ps = null;
        String url = "jdbc:mysql://xx:3306/learn?useUnicode=true&characterEncoding=utf-8&useSSL=false";
        String username = "root";
        String password = "xx";

        /**
         * 首次sink时执行，且创建到销毁只会执行一次
         * @param parameters
         * @throws Exception
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            conn = DriverManager.getConnection(url, username, password);
            conn.setAutoCommit(false);
        }

        /**
         * 数据源无数据，sink关闭时执行，且创建到销毁只会执行一次
         * @throws Exception
         */
        @Override
        public void close() throws Exception {
            if (conn != null) {
                conn.close();
            }
            if (ps != null) {
                ps.close();
            }
        }

        /**
         * 数据输出时执行，每一个数据输出时，都会执行此方法
         * @param value
         * @param context
         * @throws Exception
         */
        @Override
        public void invoke(String str, SinkFunction.Context context) throws Exception {
            String sql = "insert into demo (`test_save_point`) " +
                    "values(?)";
            ps = conn.prepareStatement(sql);
            ps.setString(1, str);
            ps.execute();
            conn.commit();
        }
    }
}

